package com.kgc.bigdata.spark.streaming

import kafka.serializer.StringDecoder
import org.apache.spark.SparkConf
import org.apache.spark.streaming.kafka.KafkaUtils
import org.apache.spark.streaming.{Seconds, StreamingContext}

/**
  * Spark Streaming整合Kafka操作：Direct方式
  */
object DirectKafkaWordCount {
  def main(args: Array[String]) {
    if (args.length < 2) {
      System.err.println(s"""
                            |Usage: DirectKafkaWordCount <brokers> <topics>
                            |  <brokers> is a list of one or more Kafka brokers
                            |  <topics> is a list of one or more kafka topics to consume from
                            |
        """.stripMargin)
      System.exit(1)
    }
    val Array(brokers, topics) = args
    val sparkConf = new SparkConf().setAppName("DirectKafkaWordCount").setMaster("local[1]")
    val ssc = new StreamingContext(sparkConf, Seconds(2))

    //我们需要消费的kafka数据的topic
    val topicsSet = topics.split(",").toSet

    // kafka的broker list地址
    val kafkaParams = Map[String, String]("metadata.broker.list" -> brokers)

    val messages = KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder](ssc, kafkaParams, topicsSet)
    messages.map(_._2)      // 取出value
      .flatMap(_.split(" ")) // 将字符串使用空格分隔
      .map(word => (word, 1))      // 每个单词映射成一个pair
      .reduceByKey(_+_)  // 根据每个key进行累加
      .print() // 打印前10个数据
    ssc.start()
    ssc.awaitTermination()
  }

}
